package fda.merge;

import java.util.LinkedList;
import java.util.List;

import fda.base.Constants;
import fda.base.Block;
import fda.base.DiskFile;
import fda.base.Memory;
import fda.base.Tuple;
import fda.schema.TableInfo;

/**
 * Class for use for Merging (or delete it and use any other if you want)
 * 
 * Note: records = tuples
 */
public class MergeEngine {

	private int inputBuffer;			// size of memory
	
	private int numberOfRuns;			// number of runs (sub. lists)
	
	private List<DiskFile> runs;
	private List<Block> buffer;	// input memory (one group i merge phase)
	private List<Tuple> output;	// output memory
	
	private final TableInfo tableInfo; // description of table currently merged
	
	public MergeEngine(TableInfo table){
		this.tableInfo = table;
	}
	
	public void setMerge(List<DiskFile> subLists) {
		Memory mem = Memory.getInstance(); 
		
		inputBuffer = Constants.MEMORY_SIZE_BLOCKS - 1;
		
		numberOfRuns = subLists.size();
		
		runs = subLists;
		
		buffer = new LinkedList<Block>();
		output = new LinkedList<Tuple>();
		
		for(int i = 0; i < Constants.MAX_TYPLES_IN_BLOCK; i++) {
			output.add(mem.getMemBlocks().get(0).getRecord(i));
		}
		
		if(inputBuffer > (numberOfRuns + 1)) {
			for(int i = 0; i < numberOfRuns; i++) {
				buffer.add(mem.getMemBlocks().get(i + 1));	// we need +1, because in memory the input is from 1,2,3...
			}
		}
		else {
			for(int i = 0; i < inputBuffer - 1; i++) {
				buffer.add(mem.getMemBlocks().get(i + 1));	// we need +1, because in memory the input is from 1,2,3...
			}
		}
	}
	
	public String merge() {
		int start = 0;			// starting list in group
		int end = 0; 			// end list in group
		int runID = 0;			// the lowest record's sub. list ID
		int cntr = 0;			// counter of the output buffer
		
		List<Tuple> records = new LinkedList<Tuple>(); // buffer for the records, which will be compared
		
		int group = 1 + (int)Math.ceil(numberOfRuns / inputBuffer);	// number of runs in groups
		
		// TODO: idea of performance increasing...need to check it (David)
		//group = (int)Math.ceil(runs.size() / inputBuffer);
		
		// go through each group of runs
		for (int j = 0; j < group; j++) {
			// temporary files
			DiskFile df;
			if (j + 1 == group)
				df = new DiskFile(Constants.PRE_SUB + "merge_" + j + "_" + getTime() + ".txt", null, tableInfo, false);
			else
				df = new DiskFile(Constants.PRE_SUB + "merge_" + j + "_" + getTime() + ".txt", null, tableInfo, true);
				
			runs.add(df);
			
			if(inputBuffer > (numberOfRuns + 1)) {
				start = j * numberOfRuns;
				end = start + numberOfRuns - 1;
			}
			else {
				start = j * inputBuffer;
				end = start + inputBuffer - 1;
			}
			
			int index = 0;
			// go through each sorted list in group
			for (int k = start; k <= end; k++) {
				if(buffer.size() == (Constants.MEMORY_SIZE_BLOCKS - 1)) {
					buffer.set(index, new Block(k, 0, runs.get(k).getBlock(0)));
					index++;
				}
				else if(buffer.size() == numberOfRuns) {
					buffer.set(index, new Block(k, 0, runs.get(k).getBlock(0)));
					index++;
				}
				else {
					buffer.add(new Block(k, 0, runs.get(k).getBlock(0)));
				}
			}
			
			// merge buffers, while there are blocks in buffer
			while (buffer.size() > 0) {
				// get first tuple from each block from buffer
				for (int l = 0; l < buffer.size(); l++) {
					records.add(buffer.get(l).getRecords().get(0));
				}
				
				Tuple firstRecord = records.get(0);
				runID = 0;
				
				// compare records to get the lowest one
				for (int m = 1; m < buffer.size(); m++) {
					// if it is lower then the last one
					if (firstRecord.getKeyValue() > records.get(m).getKeyValue()) {
						firstRecord = records.get(m);
						runID = m;
					}					
				}
				// clear for next session
				records.clear();
				
				// remove the record which was the lowest, if the record was last in the block then remove the block too
				if (buffer.get(runID).getRecords().size() > 1) {
					
					Tuple[] temp = new Tuple[buffer.get(runID).getRecords().size() - 1];
					
					for (int n = 0; n < temp.length; n++) {
						temp[n] = buffer.get(runID).getRecords().get(n + 1);				
					}
				
					// replace the the block with the modified one
					buffer.get(runID).getRecords().clear();
					for(int v = 0; v < temp.length; v++) {
						buffer.get(runID).getRecords().add(temp[v]);
					}
				}
				else {
					// if it was the last record, then remove it
					buffer.get(runID).getRecords().clear();
				}
				
				// add to output buffer
				output.set(cntr, firstRecord);
				cntr++;
				// if output buffer is full, save it to disk, then reset counter
				if(cntr > (Constants.MAX_TYPLES_IN_BLOCK - 1)) {
					runs.get(runs.size() - 1).appendTuples(output.toArray(new Tuple[output.size()]));
					cntr = 0;
				}
				
				// get next block if it is exhausted from the right sub. list
				if (buffer.get(runID).getRecords().size() == 0) {
					// are there any remaining blocks in the following sub. list
					if(runs.get(buffer.get(runID).getRunID()).getBlockCount() > (buffer.get(runID).getBlockID() + 1)) {
						Block blck = new Block(buffer.get(runID).getRunID(), buffer.get(runID).getBlockID() + 1, runs.get(buffer.get(runID).getRunID()).getBlock(buffer.get(runID).getBlockID() + 1));
						
						buffer.remove(runID);
						buffer.add(blck);
					}
					else {
						buffer.remove(runID);
					}
				}
			}
		}
		
		return runs.get(runs.size() - 1).getFilePath();
	}
	
	
	/**
	 * @return last 7 digit of current time stamp
	 */
	public static String getTime(){
		return String.valueOf(System.currentTimeMillis()).substring(6);
	}
}
